import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class LiveBusinessSqlJob {
    public static void main(String[] args) throws Exception {
        // 创建环境
        TableEnvironment tabEnv = getTableEnv();

        //2.输入表-input:映射到kafka信息队列
        createInputTable(tabEnv);

        //3.数据处理-select
        Table reportTable=handle(tabEnv);






    }

    private static Table handle(TableEnvironment tabEnv) {
        return null;
    }


    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE doris_douyin_live_business (\n" +
                "`orderId` STRING COMMENT '订单ID',\n"+
                "`eventTime` STRING COMMENT '事件时间',\n"+
                "`roomId` STRING COMMENT '房间ID',\n"+
                "`userId` STRING COMMENT '用户ID',\n"+
                "`productSku` STRING COMMENT '商品SKU',\n"+
                "`orderAmount` FLOAT COMMENT '订单金额',\n"+
                "`payAmount` STRING COMMENT '支付金额',\n"+
                "`payStatus` STRING COMMENT '支付状态',\n"+
                "`anchorId` STRING COMMENT '主播ID',\n"+
                "`roomType` STRING COMMENT '房间类型',\n"+
                "`startTime` STRING COMMENT '开始时间',\n"+
                "`endTime` STRING COMMENT '结束时间',\n"+
                "`interactId` STRING COMMENT '互动ID',\n"+
                "`interactEventType` STRING COMMENT '互动事件类型',\n"+
                "`interactType` STRING COMMENT '互动类型',\n"+
                "`content` STRING COMMENT '内容',\n"+
                "`visitId` STRING COMMENT '访问ID',\n"+
                "`enterTime` STRING COMMENT '进入时间,\n"+
                "`leaveTime` STRING COMMENT '离开时间',\n"+
                "`source` STRING COMMENT '来源',\n"+
                "`duration` INT COMMENT '时长,\n"+
                "`productName` STRING COMMENT '商品名称',\n"+
                "`price` DOUBLE COMMENT '价格,\n"+
                "`category` STRING COMMENT '分类',\n"+
                "`status` STRING COMMENT '状态',\n"+
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = 'node102:8030',  -- Doris BE 节点地址\n" +
                "  'table.identifier' = 'gd4-dyzbds.douyin_live_business',  -- Doris 库.表\n" +
                "  'username' = 'root',  -- Doris 用户名\n" +
                "  'password' = '123456'   -- Doris 密码\n" +
                ");"
        );
        //你是要把dws_traffic_page_view_window_report的数据doris_douyin_live_business里面吗
        //我要把doris_douyin_live_business 放进 gd4-dyzbds
        //嗷嗷 你这样说就是 把数据放到doris 然后用这个读
        //那直接在doris里面创表塞数据？
        //老师写过，你看看，我们组就是把数据放在Doris了 你看


    }

    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");
        configuration.setString("execution.checkpointing.interval","30 s");

        return tabEnv;
    }
}